package com.lcache.worker.service.handler;

import com.alibaba.fastjson.JSONObject;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientException;

import java.util.ArrayList;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.TimeUnit;

@Slf4j
public class MsgHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    /** 窗口时长：1 分钟 */
    private static final long WINDOW_MS = TimeUnit.MINUTES.toMillis(2);
    /** 阈值：1 分钟内超过 10 次 */
    private static final int THRESHOLD = 5;

    /** key -> 出现时间戳队列 */
    private final Map<String, Deque<Long>> keyTimestamps = new ConcurrentHashMap<>();

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) throws Exception {
        log.info("收到消息: {}", frame.text());
        JSONObject json = JSONObject.parseObject(frame.text());
        String type = json.getString("type");

        if ("login".equals(type)) {
            handleLogin(ctx, json);
        } else if ("hotkey".equals(type)) {
            handleHotKeyMessage(ctx, json);
        }
    }
    private void handleLogin(ChannelHandlerContext ctx, JSONObject json) {
        String userId = json.getString("userId");
        String username = json.getString("username");
        String password = json.getString("password");

        if (userId != null && username != null && password != null) {
            MessageDispatcher.register(userId, ctx.channel());

            // 立即响应登录成功
            sendChannelResponse(ctx.channel(), "{\"type\":\"login\",\"success\":\"true\"}");
        } else {
            sendChannelResponse(ctx.channel(),"{\"type\":\"login\",\"success\":\"false\"}");
        }
    }

    private void handleHotKeyMessage(ChannelHandlerContext ctx, JSONObject json) throws ClientException {
        String keysParam = json.getString("keys");

        long now = System.currentTimeMillis();


        String[] keys = keysParam.split(",");
        for (String rawKey : keys) {
            String key = rawKey.trim();
            if(key.isEmpty()) continue;

            // 放到后面queue, queue存储的是时间戳
            Deque<Long> deque = keyTimestamps.computeIfAbsent(key, k -> new ConcurrentLinkedDeque<>());
            deque.addLast(now);

            // 清除不符合条件的记录
            while(!deque.isEmpty() && deque.peekFirst() < now - WINDOW_MS){
                    deque.pollFirst();
            }
        }

        // 2. 收集阈值的热点Key
        List<String> hotkeys = new ArrayList<>();
        for (Map.Entry<String, Deque<Long>> entry : keyTimestamps.entrySet()) {
            if(entry.getValue().size() > THRESHOLD){
                hotkeys.add(entry.getKey());
            }
        }

        if(!hotkeys.isEmpty()){
            JSONObject out = new JSONObject();
            out.put("type", "hotkey");
            out.put("keys", String.join(",", hotkeys));
            String message = out.toJSONString();
            MessageDispatcher.dispatchToAllUser(message);
            log.info("检测到热点 keys={}，已广播消息", hotkeys);
        }

    }

    private void sendChannelResponse(Channel channel, String result) {
        channel.writeAndFlush(new TextWebSocketFrame(result));
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        log.info("连接建立: {}", ctx.channel().id());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        log.info("连接断开: {}", ctx.channel().id());
        MessageDispatcher.unregister(ctx.channel());
    }


}
